package com.mqb.quartz;

import com.alibaba.fastjson.JSON;
import com.mqb.RequestMsg;
import com.mqb.session.MsgSession;
import com.mqb.util.ByteBufUtil;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;

@DisallowConcurrentExecution
public class TimeoutMsgRetryJob implements Job {

    private static final Logger log = LoggerFactory.getLogger(TimeoutMsgRetryJob.class);


    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {

        ConcurrentHashMap<String, MsgSession.ProcessingMsg> processingMap = MsgSession.getProcessingMap();
        Iterator<MsgSession.ProcessingMsg> iterator = processingMap.values().iterator();

        LocalDateTime now = LocalDateTime.now();

        MsgSession.ProcessingMsg next;
        while (iterator.hasNext()) {
            next = iterator.next();
            long diffMillis = Duration.between(next.getDatetime(), now).toMillis();
//            if (next.getStatus() == MsgSession.INITIAL_STATUS) {
//                continue;
//            }

            if (diffMillis > 5000 && next.getStatus() == MsgSession.MIDDLE_STATUS) {
                // 超过2s没有收到 认为接收方没有收到消息，进行重发
                log.info("重发消息,类型：{} 内容：{}",next.getMsg().getClass(),JSON.toJSONString(next.getMsg()));
                MsgSession.getServerChannel().writeAndFlush(ByteBufUtil.transferWithHead(next.getMsg()));
                // 重置消息状态,重试次数+1
                MsgSession.updateStatusAfterResend(next.getMsgId());



                // debug
                RequestMsg msg = (RequestMsg) next.getMsg();
                log.info("消息被重发：重试次数 [{}], id [{}], 发送者 [{}] ,接收者 [{}], 内容 [{}]",
                       next.getRetry(), msg.getId(), msg.getSendId(), msg.getRecId(), msg.getContent());
            } else if (diffMillis > 1000 && next.getStatus() == MsgSession.INITIAL_STATUS) {
                RequestMsg msg = (RequestMsg) next.getMsg();
                // 超过1s没有收到服务器的ack，认为消息没有被服务端收到，直接标记为失败
                log.info("消息发送失败：id [{}], 发送者 [{}] ,接收者 [{}], 内容 [{}]",
                        msg.getId(), msg.getSendId(), msg.getRecId(), msg.getContent());

                // 暂时先只移除队列
                MsgSession.getProcessingMap().remove(msg.getId());
            }
        }
    }
}
